package me.sdevil507.activemq.sync;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jms.connection.ConnectionFactoryUtils;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.support.JmsUtils;
import org.springframework.stereotype.Component;

import javax.jms.*;
import java.util.UUID;

/**
 * 同步消息队列消息生产者
 * <p>
 * 同步消息队列,生产者生产一条消息需要得到一条消息消费者反馈,从而完成一次会话
 * 应用于对消息控制严格的应答场景
 *
 * @author sdevil507
 */
@Component
@Slf4j
public class SyncProducer {

    @Autowired
    @Qualifier("syncQueue")
    private Queue syncQueue;

    @Autowired
    private JmsTemplate jmsTemplate;

    /**
     * 同步模式消息发送并阻塞进程等待消息消费后返回
     *
     * @param id 传递id值
     */
    public String sendSyncMessage(String id) {
        Connection connection = null;
        Session session = null;
        try {
            connection = jmsTemplate.getConnectionFactory().createConnection();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            connection.start();
            MessageProducer producer = session.createProducer(syncQueue);
            // 需要同步模式,就必须设置JMSReplyTo,此处使用临时队列
            TemporaryQueue temporaryQueue = session.createTemporaryQueue();
            // 创建消息
            String text = "id:" + id;
            TextMessage message = session.createTextMessage(text);
            // 此处为核心,需要设置JMSReplyTo作为回执
            message.setJMSReplyTo(temporaryQueue);
            // 并且设置相关性ID
            message.setJMSCorrelationID(UUID.randomUUID().toString());
            producer.send(message);

            // 在生产者中同时创建消费者,消费消费者回复的消息
            MessageConsumer consumer = session.createConsumer(temporaryQueue);
            // 此处调用receive会阻塞线程,直到收到反馈为止
            // 超过规定时间没有返回,断掉连接(单位:ms)
            Message result = consumer.receive(60000);
            if (result instanceof TextMessage) {
                TextMessage replyText = (TextMessage) result;
                return "success" + replyText.getText();
            } else {
                return null;
            }
        } catch (JMSException e) {
            log.error("同步模式消费消息失败:", e);
            return null;
        } finally {
            JmsUtils.closeSession(session);
            ConnectionFactoryUtils.releaseConnection(connection, jmsTemplate.getConnectionFactory(), true);
        }
    }
}
